package com.g2room.bigdata.hadoop.demo.wordcount;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.util.StringTokenizer;

/**
 * 测试类
 *
 * @author zhangqingshang
 */
public class WordCount {
    static {
        System.setProperty("hadoop.home.dir", "D:\\workspace\\software\\hadoop");
    }

    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private final static IntWritable one = new IntWritable(1);
        private Text word = new Text();


        @Override
        public void map(Object key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer itr = new StringTokenizer(value.toString());
            while (itr.hasMoreTokens()) {
                word.set(itr.nextToken());
                context.write(word, one);
            }
        }
    }

    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        @Override
        public void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            result.set(sum);
            context.write(key, result);
        }
    }

    public static void main(String[] args) throws Exception {
        System.setProperty("hadoop.home.dir", "D:\\workspace\\software\\hadoop");
        // 加载Hadoop默认的一些配置。
        Configuration conf = new Configuration();
        // 不提交到yarn上，只在本地跑
        conf.set("mapreduce.framework.name", "local");
        // 初始化一个Job任务，这个Job任务需要加载hadoop的一些配置，并给这个Job命名为“word count”
        Job job = Job.getInstance(conf, "单词统计");
//        job.setJarByClass(WordCount.class);
//        job.setMapperClass(TokenizerMapper.class);
//        job.setCombinerClass(IntSumReducer.class);
//        job.setReducerClass(IntSumReducer.class);
        // 这个方法使用了WordCount.class的类加载器来寻找包含该类的Jar包，然后设置该Jar包为作业所用的Jar包。
        job.setJarByClass(WordCount.class);
        /**
         * Combiner处理类和Reducer处理类使用的是同一个类即 WordcountReducer 类
         * Combiner本质上是一个本地的Reducer方法。在集群环境中，带宽是这个集群最稀有的资源。
         * 可以先将本地map所处理的数据进行本地Reduce，然后再将结果传给集群其他节点进行处理。
         * 这样设计的好处是，可以节省集群的带宽使用率。但是，并不是所有的Combine处理过程均能使用Reducer的处理类，
         * 这需要在逻辑上考虑是否可移植！要想进行本地reduce（combine），一个必要的条件是，reduce的输入输出格式必须一样！
         */
        // 设置了该Job任务所使用的Mapper类（拆分）
        job.setMapperClass(WordcountMapper.class);
        // 设置了该Job任务所使用的Combiner类（中间结果合并），
        // Combiner接收全局所有Mapper的输出结果，对每一个maptask的输出进行局部汇总
        job.setCombinerClass(WordcountReducer.class);
        // 设置了该Job任务所使用的Reducer类（合并）
        job.setReducerClass(WordcountReducer.class);
        // 设置KEY和值的输出类型：Text类似于Java的String类型；IntWritable类型类似于Java的int类型。
        /**
         * Hadoop提供了如下内容的数据类型，这些数据类型都实现了WritableComparable接口，
         * 以便使用这些类型定义的数据可以被序列化进行网络传输和文件存储，以及进行大小比较：
         * BooleanWritable：标准布尔型数值
         * ByteWritable：单字节数值
         * DoubleWritable：双字节数
         * FloatWritable：浮点数
         * IntWritable：整型数
         * LongWritable：长整型数
         * Text：使用UTF8格式存储的文本
         * NullWritable：当<key, value>中的key或value为空时使用
         */
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 逗号分隔的多个路径 路径是在hdfs里的
        Path input = new Path("D:/workspace/code/java/open/bigdata/hadoop/src/main/resources/input");
        Path output = new Path("D:/workspace/code/java/open/bigdata/hadoop/src/main/resources/output");
        FileSystem fileSystem = output.getFileSystem(conf);
        if (fileSystem.exists(output)) {
            // true的意思是，就算output有东西，也一带删除
            fileSystem.delete(output, true);
        }
        FileInputFormat.setInputPaths(job, input);
        FileOutputFormat.setOutputPath(job, output);
        //指定我们自定义的数据分区器
        job.setPartitionerClass(WcPartitioner.class);
        //同时指定相应“分区”数量的reducetask，这里设置多少就会有多少个输出文件
        job.setNumReduceTasks(3);
        // 参数表示程序执行完，告诉我们是否执行成功
        boolean completed = job.waitForCompletion(true);
        System.exit(completed ? 0 : 1);
    }
}